﻿using org.apache.zookeeper;
using Serilog;
using StackExchange.Redis;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using static org.apache.zookeeper.Watcher;

namespace RedisDemo
{
    class Program
    {
        static void Main(string[] args)
        {
            //ConcurrencyTask concurrencyTask = new ConcurrencyTask();
            //concurrencyTask.InitRedisConnection().Wait();
            //Console.WriteLine("Redis Connection init success");

            //RedisLock(concurrencyTask);

            //concurrencyTask.DisplayNum();

            //TestRedisLockWithFencingToken();


            //ZookeeperDemo.TestConnect().GetAwaiter().GetResult();


            var outoutTemplate = "[{Timestamp:HH:mm:ss ssss} {Level:u3}] {Message:lj}{NewLine}{Exception}";
            var config = new LoggerConfiguration();
            config.MinimumLevel.Debug().WriteTo.Console(outputTemplate: outoutTemplate);
            Log.Logger = config.CreateLogger();

            //Log.Logger = new LoggerConfiguration()
            //                    .WriteTo.Console(Serilog.Events.LogEventLevel.Debug)
            //                    .CreateLogger();


            //ZkLock();

            //ZookeeperDemo.TestConnect().ConfigureAwait(false).GetAwaiter().GetResult();
            //ZookeeperDemo.TestWatch(new MyWatch()).ConfigureAwait(false);

            Enumerable.Range(1, 5).ToList().ForEach(i =>
             {
                 Task.Run(() =>
                 {

                     var lockHelper = new ZooKeeperLockHelper("localhost:5181");
                     lockHelper.OnAcquireLock += (id) =>
                     {
                         var random = new Random().Next(10);
                         Log.Debug("NodeId {@id} executing.....Sleep {@ms} ms", id, random * 1000);
                         
                         Thread.Sleep(random * 1000);
                         Log.Debug("NodeId {@id} executing success", id);

                         return Task.CompletedTask;
                     };

                     lockHelper.AcquireLock();
                    
                });

             });




            Console.ReadKey();

        }

        private static void ZkLock()
        {
            var parentPath = "/zk";
            var nodePath = "node";

            //var lockPath = "/zk/node-";

            var tasks = new List<Task>();

            List<long> sessionIds = new List<long> { 1L, 2L, 3L, 4L };


            Enumerable.Range(1, 3).ToList().ForEach(it =>
            {

                var task = Task.Run(async () =>
                {
                    var sessionId = sessionIds[it];
                    var zooKeeperHelper = new ZooKeeperHelper("localhost:5181", TimeSpan.FromSeconds(1), sessionId);
                    zooKeeperHelper.InitWatcher();
                    try
                    {
                        if (zooKeeperHelper.WaitFor(Event.KeeperState.SyncConnected, TimeSpan.FromSeconds(10)))
                        {
                            try
                            {
                                var lockPath = "/zk/node-";
                                var resultPath = await zooKeeperHelper.CreateAsync(lockPath, $"task: {it}", ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
                                var currentId = long.Parse(resultPath.Replace(lockPath, ""));
                                Log.Debug("task {@taskId} resultPath: {@path}, currentId: {id}", it, resultPath,
                                    currentId);
                                await zooKeeperHelper.WatchPath(parentPath, currentId);
                                //Console.ReadLine();

                                //var watcher = new EpemeralNodeWatcher();
                                //watcher.OnProcess += zooKeeperHelper.OnWatch;

                                //zooKeeperHelper.watchers.Add(watcher);
                                //await zooKeeperHelper.WatchPath(parentPath, currentId);

                                //var childrenResult = await zooKeeperHelper.GetChildrenAsync(parentPath, watcher);

                                //List<EpemeralNode> nodes = new List<EpemeralNode>();
                                //foreach (var c in childrenResult.Children)
                                //{
                                //    var epemeralNode = new EpemeralNode();
                                //    epemeralNode.Id = long.Parse(c.Replace(nodePath, ""));
                                //    epemeralNode.Path = parentPath + "/" + c;
                                //    nodes.Add(epemeralNode);
                                //}

                                //if (nodes.Select(node => node.Id).Min() == currentId)
                                //{
                                //    Log.Debug($"task {it} acquire lock success");
                                //    var random = new Random();
                                //    var sleepSeconds = random.Next(10);
                                //    Thread.Sleep(sleepSeconds * 1000);
                                //    zooKeeperHelper.Dispose();
                                //}
                                //else
                                //{
                                //    //var prevNodeId = currentId - 1;
                                //    //var prevPath = nodes.First(node => node.Id == prevNodeId);
                                //    //Log.Debug($"task {it} wait lock release.... prevPath {prevPath.Path}");

                                //    //var acqureLock = await zooKeeperHelper.WatchPathDeleted(prevPath.Path, TimeSpan.FromSeconds(10));
                                //    //if (acqureLock)
                                //    //{
                                //    //    Log.Debug($"task {it} retry get lock success.");
                                //    //    zooKeeperHelper.Dispose();
                                //    //}
                                //    //else
                                //    //{
                                //    //    Console.ReadLine();
                                //    //}
                                //    //Console.ReadLine();
                                //}

                                //var path = "/zk";

                                //var result = await zooKeeperHelper.GetChildrenAsync(path);
                                //result.Children.ForEach(p =>
                                //{
                                //    Log.Debug("Children: {@path}", p);
                                //});
                                //Console.ReadLine();
                            }
                            catch (Exception ex2)
                            {
                                Log.Error(ex2, "catch ex");
                            }
                        }



                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine(ex.Message);
                    }

                    //Console.ReadLine();

                });
            });
        }

        public static List<int> SLEEP_OF_SECONDS = new List<int> { 20, 0, 0, 10, 10 };

        public static void TestRedisLockWithFencingToken()
        {
            var tasks = new List<Task>();
            var mockService = new MockService();

            Enumerable.Range(1, 3).ToList().ForEach(it =>
            {
                var task = Task.Run(() =>
                {

                    var taskId = $"task_{it}";
                    var lockHelper = new RedisLockHelper();
                    var lockKey = "article:1";
                    try
                    {
                        Console.WriteLine($"-> {taskId} begin acquire redis lock.");
                        var result = lockHelper.GetLock(lockKey, TimeSpan.FromSeconds(10));
                        if (result.Success)
                        {
                            var taskInfo = new TaskInfo { TaskId = taskId, FencingToken = result.FencingToken ?? 0 };
                            Console.WriteLine($"-> {taskId} acquire lock success {taskInfo}.");
                            var sleepOfSeconds = SLEEP_OF_SECONDS[it - 1];
                            Thread.Sleep(sleepOfSeconds * 1000);
                            mockService.Save(taskInfo);
                            Console.WriteLine($"{taskInfo} Done");
                        }
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine($"{taskId} {ex.Message}");
                    }
                    finally
                    {
                        lockHelper.ReleaseLock("article:1");
                    }
                });
                tasks.Add(task);
            });

            try
            {
                Task.WaitAll(tasks.ToArray());
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                Console.WriteLine(ex.StackTrace);
            }
        }

        private static void RedisLock(ConcurrencyTask concurrencyTask)
        {
            List<Task> tasks = new List<Task>();
            Enumerable.Range(1, 3).ToList().ForEach(it =>
            {
                Task task = Task.Run(() =>
                {
                    try
                    {
                        concurrencyTask.IncreaseNum(it);
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine(ex.Message);
                    }
                });
                tasks.Add(task);

            });

            try
            {
                Task.WaitAll(tasks.ToArray());
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                Console.WriteLine(ex.StackTrace);
            }
        }
    }

    class ConcurrencyTask
    {
        public int Num;

        private IConnectionMultiplexer manager;
        private TimeSpan expiry = TimeSpan.FromSeconds(10);

        public async Task InitRedisConnection()
        {
            manager = await ConnectionMultiplexer.ConnectAsync("localhost");
        }

        public void DisplayNum()
        {
            Console.WriteLine($"Num: {Num}");
        }

        public void LuaDelKey(RedisKey key, string value)
        {
            var lua = @"if redis.call('get', @key) == @value then
                            return redis.call('del', @key)
                        else
                            return 0
                        end";
            var db = manager.GetDatabase();
            var prepare = LuaScript.Prepare(lua);
            var result = db.ScriptEvaluate(prepare, new { key = key, value = value });
        }

        public void IncreaseNum(int taskId)
        {
            //Console.WriteLine($"Task {taskId} entrant get lock");

            var redisKey = "concurrency:key:1";
            var db = manager.GetDatabase();
            var value = Guid.NewGuid().ToString().Replace("-", "");
            var lockSuccess = db.StringSet(redisKey, value, expiry, When.NotExists, CommandFlags.None);

            var start = DateTime.Now;
            while (!lockSuccess)
            {
                Thread.Sleep(50);

                Console.WriteLine($"Task {taskId} try to acquire lock {start.ToString()}");

                lockSuccess = db.StringSet(redisKey, value, expiry, When.NotExists, CommandFlags.None);

                start = start.AddMilliseconds(50);
                if (DateTime.Now.Subtract(start).TotalSeconds > 20)
                {
                    throw new InvalidOperationException($"Task {taskId} can not acquire distribute lock");
                }
            }

            if (lockSuccess)
            {
                Console.WriteLine($"{taskId} Get Lock ");
                Thread.Sleep(Program.SLEEP_OF_SECONDS[taskId - 1] * 1000);
                Num = Num + 1;
                LuaDelKey(key: redisKey, value);
                Console.WriteLine($"{taskId} Release Lock ");
            }


            //Interlocked.Increment(ref Num);
        }
    }

    public class TaskInfo
    {
        public long FencingToken { get; set; }
        public string TaskId { get; set; }

        public override string ToString()
        {
            return $"TaskId: {TaskId} - FencingToken: {FencingToken}";
        }
    }

    public class MockService
    {
        private IConnectionMultiplexer connection;
        private String fencingTokenKey = "redis_fencingToken";

        public MockService()
        {
            this.Initialize();
        }

        private void Initialize()
        {
            connection = ConnectionMultiplexer.Connect("localhost");
        }

        public void Close()
        {
            if (connection != null)
            {
                connection.Close();
            }
        }



        public void Save(TaskInfo taskInfo)
        {
            Console.WriteLine($"MockService#Save: {taskInfo}");

            var fencingToken = taskInfo.FencingToken;

            var serverFencingToken = connection.GetDatabase().StringGet(fencingTokenKey);
            Console.WriteLine($"MockService#Save: {taskInfo} - Server Fencing Token {serverFencingToken}");

            if (serverFencingToken.HasValue)
            {
                if (fencingToken == (long)serverFencingToken)
                {
                    Console.WriteLine($"MockService#Save: {taskInfo} Save Success.");
                }
                else
                {
                    Console.WriteLine($"MockService#Save: {taskInfo} Save Failure, Fencingtoken not equal server token.");
                }
            }
            else
            {
                Console.WriteLine($"{taskInfo} Save error");

            }
        }
    }

    public class RedisLockResult
    {
        public bool Success { get; set; }
        public long? FencingToken { get; set; }
    }

    public class RedisLockHelper : IDisposable
    {
        private IConnectionMultiplexer connection;
        private String value;
        private String fencingTokenKey = "redis_fencingToken";

        public RedisLockHelper()
        {
            value = Guid.NewGuid().ToString();
            this.Initialize();
        }

        public RedisLockResult GetLock(String key, TimeSpan expiry)
        {
            var db = connection.GetDatabase();
            var lockSuccess = db.StringSet(key, value, expiry, When.NotExists, CommandFlags.None);

            var start = DateTime.Now;
            while (!lockSuccess)
            {
                Thread.Sleep(50);
                lockSuccess = db.StringSet(key, value, expiry, When.NotExists, CommandFlags.None);
                start = start.AddMilliseconds(50);
                if (DateTime.Now.Subtract(start).TotalSeconds > 20)
                {
                    throw new InvalidOperationException($"can not acquire distributed lock");
                }
            }

            var result = new RedisLockResult()
            {
                Success = lockSuccess
            };

            if (lockSuccess)
            {
                var fencingToken = db.StringIncrement(fencingTokenKey, 1, CommandFlags.None);
                result.FencingToken = fencingToken;
            }

            return result;
        }

        public void ReleaseLock(RedisKey key)
        {
            var lua = @"if redis.call('get', @key) == @value then
                            return redis.call('del', @key)
                        else
                            return 0
                        end";
            var db = connection.GetDatabase();
            var prepare = LuaScript.Prepare(lua);
            var result = db.ScriptEvaluate(prepare, new { key = key, value = value });
        }

        private void Initialize()
        {
            connection = ConnectionMultiplexer.Connect("localhost");
        }

        public void Dispose()
        {
            if (connection != null)
            {
                connection.Dispose();
            }
        }
    }
}
